package com.learn.serialier;

import com.learn.bean.ToKafkaSinkPojo;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.nio.charset.StandardCharsets;

public class SerializationTest implements KafkaRecordSerializationSchema<ToKafkaSinkPojo> {

    String topic;


    public SerializationTest() {

    }

    public SerializationTest(String topic) {

        this.topic = topic;
    }


    @Override
    public ProducerRecord<byte[], byte[]> serialize(ToKafkaSinkPojo element, KafkaSinkContext context, Long timestamp) {

        String key = element.getKey();

        String value = element.getValue();

        ProducerRecord producerRecord = new ProducerRecord<byte[], byte[]>(topic, null, null, key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8));

        return producerRecord;
    }


}
